package com.hr.reactive.flux;


import com.hr.reactive.common.Utils;
import com.hr.reactive.mono.s01.sub.SubscriberImpl;
import reactor.core.publisher.Flux;

/**
 * @author huran
 */
public class Lec09FluxCreateDownsteamDemand {
    public static void main(String[] args) {
//        var sub = new SubscriberImpl();
//        Flux.<String>create(fluxSink -> {
//            for (int i = 0; i < 10; i++) {
//                var name = Utils.faker().name().firstName();
//                System.out.println("===> " + name);
//                fluxSink.next(name);
//            }
//            fluxSink.complete();
//        }).subscribe(sub);
//        Utils.sleepSeconds(2000);
//        sub.getSubscription().request(2);
//        Utils.sleepSeconds(2000);
//        sub.getSubscription().request(2);
//        Utils.sleepSeconds(2000);
//        sub.getSubscription().cancel();
        productDemo();
    }

    //按需生产
    private static void productDemo() {
        var sub = new SubscriberImpl();
        Flux.<String>create(stringFluxSink -> {
            stringFluxSink.onRequest(req -> {
                for (int i = 0; i < req && !stringFluxSink.isCancelled(); i++) {
                    var name = Utils.faker().name().firstName();
                    System.out.println("===> " + name);
                    stringFluxSink.next(name);
                }
            });

        }).subscribe(sub);

        Utils.sleepSeconds(2000);
        sub.getSubscription().request(2);
        Utils.sleepSeconds(2000);
        sub.getSubscription().request(3);
        Utils.sleepSeconds(2000);
        sub.getSubscription().cancel();
        sub.getSubscription().request(3);
    }
}
